package com.atguigu.day03;

import com.atguigu.day02.Example2;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RichAggregateFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Tuple2;

import java.util.Calendar;

public class Example2_1 {
    public static void main(String[] args)throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<com.atguigu.day02.Example2.Event> streamSource = env.addSource(new com.atguigu.day02.Example2.MySource());
        KeyedStream<com.atguigu.day02.Example2.Event, String> keyedStream = streamSource.keyBy(x -> x.name);
        WindowedStream<com.atguigu.day02.Example2.Event, String, TimeWindow> windowedStream = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(5)));
        SingleOutputStreamOperator<Example1.UserViewCountPerWindow> singleOutputStreamOperator = windowedStream.aggregate(new CountACC());
        singleOutputStreamOperator.print();
        env.execute();
    }
public static class CountACC implements AggregateFunction<Example2.Event, Tuple4<String,Long,Long,Long>, Example1.UserViewCountPerWindow> {

    @Override
    public Tuple4<String, Long, Long, Long> createAccumulator() {
        return new Tuple4<>(null,0L, 0L,0L);
    }

    @Override
    public Tuple4<String, Long, Long, Long> add(Example2.Event event, Tuple4<String, Long, Long, Long> stringLongLongLongTuple4) {
        return new Tuple4<>(event.name,stringLongLongLongTuple4.f1+1, Calendar.getInstance().getTimeInMillis()-Calendar.getInstance().getTimeInMillis()%5000, stringLongLongLongTuple4.f3);
    }

    @Override
    public Example1.UserViewCountPerWindow getResult(Tuple4<String, Long, Long, Long> stringLongLongLongTuple4) {
        return new Example1.UserViewCountPerWindow(
                stringLongLongLongTuple4.f0,
                stringLongLongLongTuple4.f1,
                stringLongLongLongTuple4.f2,
                Calendar.getInstance().getTimeInMillis()
        );
    }

    @Override
    public Tuple4<String, Long, Long, Long> merge(Tuple4<String, Long, Long, Long> stringLongLongLongTuple4, Tuple4<String, Long, Long, Long> acc1) {
        return null;
    }
}

}
